package com.junmoo.www.config.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.*;
import org.springframework.stereotype.Component;

@Component
@IntegrationComponentScan
@Slf4j
public class MqttConfig {
    /**
     * 订阅的bean名称
     */
    public static final String CHANNEL_NAME_IN = "mqttInboundChannel";
    /**
     * 发布的bean名称
     */
    public static final String CHANNEL_NAME_OUT = "mqttOutboundChannel";


    @Value("${spring.mqtt.inbound.url}")
    private String inboundHostUrl;

    @Value("${spring.mqtt.inbound.clientId}")
    private String inboundClientId;

    @Value("${spring.mqtt.inbound.username}")
    private String inboundUserName;

    @Value("${spring.mqtt.inbound.password}")
    private String inboundPassWord;

    @Value("${spring.mqtt.inbound.topics}")
    private String inboundTopics;

    @Value("${spring.mqtt.outbound.urls}")
    private String outboundHostUrls;

    @Value("${spring.mqtt.outbound.clientId}")
    private String outboundClientId;

    @Value("${spring.mqtt.outbound.username}")
    private String outboundUserName;

    @Value("${spring.mqtt.outbound.password}")
    private String outboundPassWord;

    @Value("${spring.mqtt.outbound.topic}")
    private String outboundTopic;



    @Bean
    public MqttConnectOptions mqttConnectOptions(){
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setServerURIs( outboundHostUrls.split(",") );
        mqttConnectOptions.setUserName(outboundUserName);
        mqttConnectOptions.setPassword(outboundPassWord.toCharArray());
        mqttConnectOptions.setKeepAliveInterval(60);
        return mqttConnectOptions;
    }
//--------------- 发送通道 -------------------------------------------
    @Bean
    public MqttPahoClientFactory mqttClientFactory(MqttConnectOptions mqttConnectOptions){
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConnectOptions);
        return factory;
    }

    //发送通道
    @Bean(name = CHANNEL_NAME_OUT)
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = CHANNEL_NAME_OUT)
    public MessageHandler MqttOutbound(MqttPahoClientFactory mqttClientFactory){
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(outboundClientId,mqttClientFactory);
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(outboundTopic);
        return messageHandler;
    }
//----------------------接收通道---------------------------------------------------------------------------------------
    //接收通道
    @Bean(name = CHANNEL_NAME_IN)
    public MessageChannel mqttInboundChannel() {
        return new DirectChannel();
    }

    //配置client,监听的topic
    @Bean
    public MessageProducer inbound(MqttPahoClientFactory mqttClientFactory,MessageChannel mqttInboundChannel) {
//        MqttPahoMessageDrivenChannelAdapter adapter =
//                new MqttPahoMessageDrivenChannelAdapter(inboundHostUrl,inboundClientId,mqttClientFactory,inboundTopic.split(","));
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(inboundClientId,mqttClientFactory,inboundTopics.split(","));
        adapter.setCompletionTimeout(500000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInboundChannel);
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = CHANNEL_NAME_IN)
    public MessageHandler handler() {
        return new MessageHandler() {
            // from device ,  发送到kakfa ;  from client , 等待设备消费消息
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                log.info("===== "+this.getClass().getName()+"."+Thread.currentThread().getStackTrace()[1].getMethodName());
                // 消息from device , from client
                //Header 包含：
                //      mqtt_receivedRetained : true/false
                //      mqtt_id:
                //      mqtt_duplicate : true/false
                //      id :
                //      mqtt_receivedTopic : "system/r1"
                //      mqtt_receivedQos
                //      timestamp :
                MessageHeaders messageHeaders = message.getHeaders();
                String messageId = messageHeaders.getId().toString();
                String ClientId = messageHeaders.get("mqtt_id").toString();
                String key = ClientId+"-"+messageId;
                String topic =String.valueOf(messageHeaders.get("mqtt_receivedTopic")).toLowerCase();
                String payload = String.valueOf(message.getPayload() );
            }
        };
    }

}
